package com.airikeji.hr.rabbitmq.config;

import cn.hutool.json.JSONUtil;
import com.airikeji.hr.common.SpringBeanUtils;
import com.airikeji.hr.common.StringPool;
import com.airikeji.hr.domain.Exchanges;
import com.airikeji.hr.domain.MessageConfigDTO;
import com.airikeji.hr.domain.Queues;
import com.airikeji.hr.service.MessageConfigService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * rabbitmq 消息队列和交换机 配置文件
 * @Author huangrui
 * @Date 2021/7/13 16:14
 * @Version 1.0
 */
@Slf4j
@Data
@Component
public class RabbitMqProperties implements InitializingBean{

    @Autowired
    private MessageConfigService messageConfigService;

    @Override
    public void afterPropertiesSet() throws Exception {
        String messageConfig = messageConfigService.querySystemConfig();
        MessageConfigDTO messageConfigDTO = JSONUtil.toBean(messageConfig, MessageConfigDTO.class);
        this.exchanges = messageConfigDTO.getExchanges();
        this.queues = messageConfigDTO.getQueues();
    }

    private List<Exchanges> exchanges = null;
    private List<Queues> queues = null;

    public Exchanges getExchangeConfig(String name) {
        Map<String, Exchanges> collect = exchanges.stream().collect(Collectors.toMap(e -> e.getName(), e -> e));
        return collect.get(name);
    }

    /**
     * 动态创建交换机
     *
     * @return
     */
    @Bean
    @Transactional(rollbackFor = Exception.class)
    public Object createExchange() {
        List<Exchanges> exchanges = getExchanges();
        if (!CollectionUtils.isEmpty(exchanges)) {
            exchanges.forEach(e -> {
                // 声明交换机
                Exchange exchange = null;
                switch (e.getType()) {
                    case DIRECT:
                        exchange = new DirectExchange(e.getName());
                        break;
                    case TOPIC:
                        exchange = new TopicExchange(e.getName());
                        break;
                    case HEADERS:
                        exchange = new HeadersExchange(e.getName());
                        break;
                    case FANOUT:
                        exchange = new FanoutExchange(e.getName());
                        break;
                    case CUSTOM:
                        exchange = new CustomExchange(e.getName(), e.getCustomType(), true, false, e.getArguments());
                        break;
                    default:
                        log.warn("未指定[{}]交换机配置!", e.getName());
                        break;
                }
                // 将交换机注册到spring bean工厂 让spring实现交换机的管理
                if (exchange != null) {
                    SpringBeanUtils.registerBean(e.getName(), exchange);
                }
            });
        }

        return null;
    }

    /**
     * 动态绑定队列和交换机
     *
     * @return
     */
    @Bean
    @Transactional(rollbackFor = Exception.class)
    public Object bindingQueueToExchange() {
        List<Queues> queues = getQueues();
        if (!CollectionUtils.isEmpty(queues)) {
            queues.forEach(q -> {
                // 创建队列
                Queue queue = new Queue(q.getName(), q.getDurable(),
                        q.getExclusive(), q.getAutoDelete(), q.getArgs());

                // 注入队列bean
                SpringBeanUtils.registerBean(q.getName(), queue);

                // 获取队列绑定交换机名
                List<String> exchangeNameList;
                if (q.getExchangeName().indexOf(StringPool.COMMA) != -1) {
                    String[] split = q.getExchangeName().split(StringPool.COMMA);
                    exchangeNameList = Arrays.asList(split);
                } else {
                    exchangeNameList = Arrays.asList(q.getExchangeName());
                }
                exchangeNameList.forEach(name -> {
                    // 获取交换机配置参数
                    Exchanges exchanges = getExchangeConfig(name);
                    Binding binding = bindingBuilder(queue, q, exchanges);

                    // 将绑定关系注册到spring bean工厂 让spring实现绑定关系的管理
                    if (binding != null) {
                        log.debug("队列 [{}] 绑定交换机 [{}] 成功!", q.getName(), exchanges.getName());
                        SpringBeanUtils.registerBean(q.getName() + StringPool.DASH + name, binding);
                    }
                });

            });
        }
        return null;
    }

    public Binding bindingBuilder(Queue queue, Queues q, Exchanges exchanges) {
        // 声明绑定关系
        Binding binding = null;

        // 根据不同的交换机模式 获取不同的交换机对象（注意：刚才注册时使用的是父类Exchange，这里获取的时候将类型获取成相应的子类）生成不同的绑定规则
        switch (exchanges.getType()) {
            case TOPIC:
                binding = BindingBuilder.bind(queue)
                        .to(SpringBeanUtils.getBean(exchanges.getName(), TopicExchange.class))
                        .with(q.getRoutingKey());
                break;
            case DIRECT:
                binding = BindingBuilder.bind(queue)
                        .to(SpringBeanUtils.getBean(exchanges.getName(), DirectExchange.class))
                        .with(q.getRoutingKey());
                break;
            case HEADERS:
                if (q.getWhereAll()) {
                    binding = BindingBuilder.bind(queue)
                            .to(SpringBeanUtils.getBean(exchanges.getName(), HeadersExchange.class))
                            .whereAll(q.getHeaders()).match();
                } else {
                    binding = BindingBuilder.bind(queue)
                            .to(SpringBeanUtils.getBean(exchanges.getName(), HeadersExchange.class))
                            .whereAny(q.getHeaders()).match();
                }
                break;
            case FANOUT:
                binding = BindingBuilder.bind(queue)
                        .to(SpringBeanUtils.getBean(exchanges.getName(), FanoutExchange.class));
                break;
            case CUSTOM:
                binding = BindingBuilder.bind(queue)
                        .to(SpringBeanUtils.getBean(exchanges.getName(), CustomExchange.class))
                        .with(q.getRoutingKey()).noargs();
                break;
            default:
                log.warn("未指定 [{}] 队列配置!", q.getName());
                break;
        }

        return binding;
    }
}

